package com.lhl.kafka.kafkaconsumer;

import junit.framework.JUnit4TestAdapter;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @author LHL
 * @date 2019/4/11
 */
@SpringBootTest
@RunWith(SpringRunner.class)
public class KafkaProducer {

    @Autowired
    private KafkaTemplate kafkaTemplate;

   @Test
    public void testBatch() throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            if (i==6){
                Thread.sleep(1000*10);
            }
            ListenableFuture<SendResult<String,String>> demo1 = kafkaTemplate.send("demo1", "dataNum-" + i);
            int finalI = i;
            demo1.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("=============失败"+throwable.getCause()+ finalI);
                }

                @Override
                public void onSuccess(SendResult<String, String> stringStringSendResult) {
                    System.out.println("=====================成功");
                }
            });
        }
    }
}
